#!usr/bin/env python  
# -*- coding:utf-8 _*-
""" 
@author:robot
@file: mq01_activemq.py 
@version:
@time: 2022/03/07 
"""
import time
import stomp
import json
from TapClientConnector.connector.tap_connector00_base import *

pk = 'pk'
active_split = '<<<>>>'


class TapActiveMQConnector(TAPConnectorBase):
    pdk = None

    def __init__(self, kwargs):
        super(TapActiveMQConnector, self).__init__(kwargs)
        brokerURL = kwargs.get('pythonURL')
        host_and_port = brokerURL.split("//")[1].split(":")
        host = host_and_port[0]
        port = host_and_port[1]
        mqUserName = kwargs.get('mqUserName')
        mqPassword = kwargs.get('mqPassword')
        self.pdk = kwargs.get('pdk')
        self.op = 'TAPDATA_SYNC_OP'
        self.sync_data = 'TAPDATA_SYNC_DATA'
        self.op_insert = 'INSERT'
        self.op_update = 'UPDATE'
        self.op_delete = 'DELETE'

        if self.pdk:
            self.op = 'mqOp'
            self.sync_data = 'sync_data'
            self.op_insert = 'insert'
            self.op_update = 'update'
            self.op_delete = 'delete'
        self.FORMAT_DB_INFO = {
            'host': host,
            'port': port,
            'user': mqUserName,
            'password': mqPassword,
            'op': self.op,
            'sync_data': self.sync_data,
            'op_insert': self.op_insert,
            'op_update': self.op_update,
            'op_delete': self.op_delete,
            'pdk': self.pdk
        }
        self.active_data = {}
        self.active_json_data = []
        self.timeout = 3600
        con = ActiveMQProducer(self.FORMAT_DB_INFO)

    def __activemq_transform(self, des, pk=pk):
        for column in self.active_data[des]:
            try:
                op = column[self.op]
                data = column[self.sync_data]
                if op == self.op_insert:
                    self.active_json_data.append(data)
                if op == self.op_update:
                    for old in self.active_json_data:
                        try:
                            if old[pk] == data[pk]:
                                # for new_k, new_value in data.items():
                                #     old[new_k] = new_value
                                self.active_json_data.remove(old)
                                self.active_json_data.append(data)
                        except:
                            logger.warning('active 数据没有pk预设主键，查询整理结果无效')
                if op == self.op_delete:
                    for old in self.active_json_data:
                        try:
                            if old[pk] == data[pk]:
                                self.active_json_data.remove(old)
                        except:
                            logger.warning('active 数据没有pk预设主键，查询整理结果无效')
            except:
                logger.warning("can't transform")
                self.active_json_data = self.active_data[des]

    def activemq_find(self, table, timeout=10):
        self.FORMAT_DB_INFO.update({'des': table})
        con = ActiveMQConsumer(self.FORMAT_DB_INFO)
        return con.query(timeout)

    def __activemq_find(self, table, timeout):
        self.FORMAT_DB_INFO.update({'des': table})
        con = ActiveMQConsumer(self.FORMAT_DB_INFO)
        return con.query(timeout)

    @connector_query_logger
    def _query(self, table):
        where = None
        # 用于where筛选
        if active_split in table:
            _table = table.split()
            table = _table[0]
            where = _table[1]
        self.active_data[f'{table}'] = []
        res = self.__activemq_find(table, timeout=self.timeout)
        if not res:
            logger.fatal(f'activemq {table} consumer timeout {self.timeout} ms')
        self.active_data[f'{table}'].extend(res)
        self.__activemq_transform(des=table)
        res = self.active_json_data
        self.active_json_data = []
        self.active_data = {}
        if where:
            for one in res:
                if one[pk] == where:
                    return one
            return []
        return res

    def _exec(self, data):
        try:
            operate = data.get('operate')
            topic_queue = data.get('topic_queue')
            topic_queue_data = data.get('data')
        except:
            operate = 'insert'
            topic_queue = data
            topic_queue_data = {
                "pk": 1,
                "string": "string",
                "number": 1000,
                'boolean': True,
                'Array': [1, 2, 3],
                'object': {'test': 'test'}
            }
        if operate == 'insert':
            json_data = json.dumps(topic_queue_data)
            self.FORMAT_DB_INFO.update({'des': topic_queue})
            con = ActiveMQProducer(self.FORMAT_DB_INFO)
            con.insert(json_data)
        elif operate == 'update':
            json_data = json.dumps(topic_queue_data)
            self.FORMAT_DB_INFO.update({'des': topic_queue})
            con = ActiveMQProducer(self.FORMAT_DB_INFO)
            con.update(json_data)
        elif operate == 'delete':
            json_data = json.dumps(topic_queue_data)
            self.FORMAT_DB_INFO.update({'des': topic_queue})
            con = ActiveMQProducer(self.FORMAT_DB_INFO)
            con.delete(json_data)

    def drop(self, table):
        self.__activemq_find(table, 5)

    def generate_insert(self, insert, table=None):
        return {
            'operate': 'insert',
            'topic_queue': table,
            'data': insert,
        }

    @generate_logger
    def generate_update(self, update, where, table=None):
        if table:
            self.CURRENT_TABLE = table
        return {
            'operate': 'update',
            'data': (where, update),
            'topic_queue': self.CURRENT_TABLE
        }

    @generate_logger
    def generate_delete(self, where, table=None):
        if table:
            self.CURRENT_TABLE = table
        return {
            'operate': 'delete',
            'data': where,
            'topic_queue': self.CURRENT_TABLE
        }

    @generate_logger
    def generate_drop(self, table=None):
        if table:
            self.CURRENT_TABLE = table
        return {
            'operate': 'drop',
            'topic_queue': [self.CURRENT_TABLE]
        }

    @generate_logger
    def generate_create(self, table_column, table=None):
        if table:
            self.CURRENT_TABLE = table
        return self.generate_insert(table_column, table=self.CURRENT_TABLE)

    @generate_logger
    def generate_select(self, table=None, where=None, order=''):
        self.timeout = 20
        if table:
            self.CURRENT_TABLE = table
        if where:
            return f'{self.CURRENT_TABLE}{active_split}{where.get(pk)}'
        return self.CURRENT_TABLE


def connect_and_subscribe(connector, connect_info):
    connector.connect(connect_info.get('user'),
                      connect_info.get('password'),
                      wait=True)
    connector.subscribe(destination=f'{connect_info.get("des")}', id=1, ack='client ')


class ActiveMQProducer:

    def __init__(self, connect_info):
        host = connect_info.get('host')
        port = connect_info.get('port')
        username = connect_info.get('user')
        password = connect_info.get('password')
        self.des = connect_info.get('des')
        self.op = connect_info.get('op')
        self.sync_data = connect_info.get('sync_data')
        self.op_insert = connect_info.get('op_insert')
        self.op_update = connect_info.get('op_update')
        self.op_delete = connect_info.get('op_delete')
        self.conn = stomp.Connection([(host, int(port))])
        self.conn.connect(username, password, wait=True)

    def insert(self, message):
        self.conn.send(f'{self.des}', message, headers={self.op: self.op_insert, "amq-msg-type": "text"})

    def update(self, message):
        self.conn.send(f'{self.des}', message, headers={self.op: self.op_update, "amq-msg-type": "text"})

    def delete(self, message):
        self.conn.send(f'{self.des}', message, headers={self.op: self.op_delete, "amq-msg-type": "text"})


class MyListener(stomp.ConnectionListener):
    data = []

    def __init__(self, conn):
        self.conn = conn

    def on_error(self, frame):
        logger.fatal(f'received an error {frame.body}')

    def on_message(self, frame):
        logger.info(f'received  {frame.headers}')
        logger.info(f'received  {frame.body}')
        consume_message = {'TAPDATA_SYNC_OP': frame.headers.get('TAPDATA_SYNC_OP'),
                           'TAPDATA_SYNC_DATA': json.loads(frame.body)}
        self.data.append(consume_message)


class MyListenerPDK(stomp.ConnectionListener):
    data = []

    def __init__(self, conn):
        self.conn = conn

    def on_error(self, frame):
        logger.fatal(f'received an error {frame.body}')

    def on_message(self, frame):
        logger.info(f'received  {frame.headers}')
        logger.info(f'received  {frame.body}')
        consume_message = {'mqOp': frame.headers.get('mqOp'),
                           'sync_data': json.loads(frame.body)}
        self.data.append(consume_message)


class ActiveMQConsumer:
    def __init__(self, connect_info):
        self.connect_info = connect_info
        host = connect_info.get('host')
        port = connect_info.get('port')
        username = connect_info.get('user')
        password = connect_info.get('password')
        self.des = connect_info.get('des')
        self.conn = stomp.Connection([(host, int(port))], heartbeats=(4000, 4000))
        pdk = connect_info.get('pdk')
        if pdk:
            self.listener = MyListenerPDK(self.conn)
        else:
            self.listener = MyListener(self.conn)
        self.conn.set_listener('python', self.listener)
        self.data = self.listener.data

    def query(self, timeout):
        connect_and_subscribe(self.conn, self.connect_info)
        time.sleep(timeout)
        self.conn.disconnect()
        return self.data

    def __del__(self):
        pass


if __name__ == '__main__':
    # con_info = {
    #     'host': '192.168.1.183',
    #     'port': '61613',
    #     'user': 'admin',
    #     'password': 'admin',
    #     'des': '/topic/zsj'
    # }
    # con = ActiveMQConsumer(con_info)
    # con.query()
    from TapDataSourceInfo.db_info_pdk import ActiveMQ

    t = TapActiveMQConnector(ActiveMQ.AUTO_ACTIVEMQ_SOURCE)
    t.timeout = 5
    print(t._query('/queue/auto_dummy'))
